package cn.xuexiyuan.flinkstudy.watermark;

import cn.xuexiyuan.flinkstudy.entity.Order;
import cn.xuexiyuan.flinkstudy.test.DataFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * @Description: 基与事件时间的窗口计算 + watermark 解决一定程度数据乱序/延迟到达的问题
 *
 * @Author 左龙龙
 * @Date 21-3-25
 * @Version 1.0
 **/
public class WatermarkDemo01 {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Order> orderDS = env.addSource(new SourceFunction<Order>() {
            private boolean running = true;
            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                while(running){
                    Order order = DataFactory.createRandomOrder(10, 5);
                    ctx.collect(order);

                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });

        // 2.transformation
        // 在最新版本中默认的就是 EventTime
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        SingleOutputStreamOperator<Order> orderDSWithWatermark = orderDS.assignTimestampsAndWatermarks(
                // 设置 maxOutOfOrderness 最大无序度/最大允许的延迟/乱序时间
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        // 指定事件时间列
                        .withTimestampAssigner((order, timestamp) -> order.getOrder_time())
        );

        SingleOutputStreamOperator<Order> result = orderDSWithWatermark
                .keyBy(Order::getUser_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum("payment_money");


        // 3.sink
        result.print();

        // 4.excute
        env.execute();

    }

}
